#ifndef __M_CONSUMER_H__
#define __M_CONSUMER_H__
#include"../mqcommon/mq_msg.pb.h"
#include"../mqcommon/helper.hpp"
#include"../mqcommon/logger.hpp"
#include<functional>
#include<mutex>
#include<memory>

namespace mq{
    using ConsumerCallback = std::function<void(const std::string&,const BasicProperty* bp,const std::string&)>;
    struct Consumer{
        using ptr = std::shared_ptr<Consumer>;
        std::string tag;
        std::string qname;
        bool auto_ack;
        ConsumerCallback cb;
        Consumer(){}
        Consumer(const std::string& ctag,const std::string& cqname,bool cack,const ConsumerCallback& ccb)
        :tag(ctag),
        qname(cqname),
        auto_ack(cack),
        cb(ccb){}
    };

    class QueueConsumer{
        public:
            using ptr = std::shared_ptr<QueueConsumer>;
            QueueConsumer(const std::string& qname):_qname(qname){}

            Consumer::ptr create(const std::string& ctag, bool cack,const ConsumerCallback& cb){
                std::unique_lock<std::mutex> lock(_mutex);
                for(auto it = _consumers.begin();it != _consumers.end();++it){
                    if((*it)->tag == ctag) 
                        return (*it);
                }
                Consumer::ptr csp = std::make_shared<Consumer>(ctag, _qname,cack,cb);
                _consumers.push_back(csp);
                return csp;
            }
            void remove(const std::string& ctag){
                std::unique_lock<std::mutex> lock(_mutex);
                for(auto it = _consumers.begin();it != _consumers.end();++it){
                    if((*it)->tag == ctag) {
                        _consumers.erase(it);
                        return;
                    }

                }
            }

            Consumer::ptr chose(){
                if(isEmpty()) return Consumer::ptr();
                std::unique_lock<std::mutex> lock(_mutex);
                Consumer::ptr csp = _consumers[_rr_sep];
                _rr_sep = (++_rr_sep) % _consumers.size();
                return csp;
            }

            bool isEmpty(){
                std::unique_lock<std::mutex> lock(_mutex);
                return _consumers.size() == 0;
            }

            bool exists(const std::string& ctag){
                std::unique_lock<std::mutex> lock(_mutex);
                for(auto it = _consumers.begin();it != _consumers.end();++it){
                    if((*it)->tag == ctag) 
                        return true;
                }
                return false;
            }
            void clear(){
                std::unique_lock<std::mutex> lock(_mutex);
                _consumers.clear();
                _rr_sep == 0;
            }   
        private:
            std::string _qname;
            std::mutex _mutex;
            uint64_t _rr_sep;
            std::vector<Consumer::ptr> _consumers;
    };

    class ConsumerManager{
        public:
        using ptr = std::shared_ptr<ConsumerManager>;
            ConsumerManager(){
                LOG_DEBUGM("consumerManager init...");

            }
            bool initQueueConsumer(const std::string& qname){
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it != _qconsumers.end()) return true;
                auto qconsumer = std::make_shared<QueueConsumer>(qname);
                _qconsumers.insert(std::make_pair(qname,qconsumer));
                return true;
            }
            void destoryQueueConsumer(const std::string& qname){
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end()) return ;
                _qconsumers.erase(it);
            }
            Consumer::ptr createConsumer(const std::string& ctag, const std::string& qname,bool cack,
                const ConsumerCallback& cb){
                QueueConsumer::ptr qcp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _qconsumers.find(qname);
                    if(it == _qconsumers.end()) {
                        LOG_DEBUGM("没有找到存储队列%s 消费者管理句柄",qname.c_str());
                        return Consumer::ptr();
                    }
                    qcp = it->second;
                }
                return qcp->create(ctag,cack,cb);
            }

            void removeConsumer(const std::string& ctag, const std::string& qname){
                QueueConsumer::ptr qcp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _qconsumers.find(qname);
                    if(it == _qconsumers.end()) {
                        return ;
                    }
                    qcp = it->second;
                }
                qcp->remove(ctag);
            }

            Consumer::ptr choose(const std::string& qname){
                QueueConsumer::ptr qcp;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _qconsumers.find(qname);
                    if(it == _qconsumers.end()) {
                        LOG_DEBUGM("没有找到存储队列%s 消费者管理句柄",qname.c_str());
                        return Consumer::ptr();
                    }
                    qcp = it->second;
                }
                return qcp->chose();
            }

            bool isEmptyQueueConsumer(const std::string& qname){
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end()) {
                    LOG_DEBUGM("没有找到存储队列%s 消费者管理句柄",qname.c_str());
                    return true;
                }
                return it->second->isEmpty();
            }

            bool exists(const std::string& qname,const std::string& ctag){
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end()) return false;

                return it->second->exists(ctag);
            }
            void clear(){
                std::unique_lock<std::mutex> lock(_mutex);
                _qconsumers.clear();
            }
        private:
            std::mutex _mutex;
            std::unordered_map<std::string,QueueConsumer::ptr> _qconsumers;
    };
}

#endif